 1.Kafka高级特性解析之稳定性中控制器
   
   Kafka集群包含若干个broker，broker.id指定broker的编号，编号不要重复。
   Kafka集群上创建的主题，包含若干个分区。
   每个分区包含若干个副本，副本因子包括了Follower副本和Leader副本。
   副本又分为ISR（同步副本分区）和OSR（非同步副本分区）。
   控制器就是一个broker。
   控制器除了一般broker的功能，还负责Leader分区的选举。
   1).broker选举
   集群里第一个启动的broker在Zookeeper中创建临时节点<KafkaZkChroot>/controller 。
   其他broker在该控制器节点创建Zookeeper watch对象，使用Zookeeper的监听机制接收该节点的变更。
   即：Kafka通过Zookeeper的分布式锁特性选举集群控制器。
   下图中，节点/myKafka/controller 是一个zookeeper临时节点，其中"brokerid":0 ，表示当前控制器是
broker.id为0的broker。
   [zk: localhost:2181(CONNECTED) 0] get /myKafka/controller
{"version":1,"brokerid":0,"timestamp":"1641390892554"}
cZxid = 0x1c00000004
ctime = Wed Jan 05 21:54:52 CST 2022
mZxid = 0x1c00000004
mtime = Wed Jan 05 21:54:52 CST 2022
pZxid = 0x1c00000004
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x3000000ddbc0001
dataLength = 54
numChildren = 0
[zk: localhost:2181(CONNECTED) 1] 
   每个新选出的控制器通过 Zookeeper 的条件递增操作获得⼀个全新的、数值更大的 controller epoch。其他
broker 在知道当前 controller epoch 后，如果收到由控制器发出的包含较旧epoch 的消息，就会忽略它们，以防止
“脑裂”。
   比如当一个Leader副本分区所在的broker宕机，需要选举新的Leader副本分区，有可能两个具有不同纪元数字的
控制器都选举了新的Leader副本分区，如果选举出来的Leader副本分区不一样，听谁的？脑裂了。有了纪元数字，直
接使用纪元数字最新的控制器结果。
   [zk: localhost:2181(CONNECTED) 1] get /myKafka/controller_epoch
2270
cZxid = 0xd0000001c
ctime = Fri Dec 17 00:39:03 CST 2021
mZxid = 0x1c00000006
mtime = Wed Jan 05 21:54:52 CST 2022
pZxid = 0xd0000001c
cversion = 0
dataVersion = 2269
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 4
numChildren = 0
[zk: localhost:2181(CONNECTED) 2]
   当控制器发现一个 broker 已经离开集群，那些失去Leader副本分区的Follower分区需要一个新Leader（这些分
区的首领刚好是在这个 broker 上）。
   (1).控制器需要知道哪个broker宕机了？
   (2).控制器需要知道宕机的broker上负责的时候哪些分区的Leader副本分区？
   下图中， <KafkaChroot>/brokers/ids/0 保存该broker的信息，此节点为临时节点，如果broker节点宕机，该
节点丢失。
   集群控制器负责监听ids 节点，一旦节点子节点发送变化，集群控制器得到通知。
   [zk: localhost:2181(CONNECTED) 2] get /myKafka/brokers/ids/0
{"listener_security_protocol_map":{"PLAINTEXT":"PLAINTEXT"},
"endpoints":["PLAINTEXT://linux121:9092"],
"jmx_port":-1,"host":"linux121","timestamp":"1641390892874","port":9092,"version":4}
cZxid = 0x1c00000009
ctime = Wed Jan 05 21:54:53 CST 2022
mZxid = 0x1c00000009
mtime = Wed Jan 05 21:54:53 CST 2022
pZxid = 0x1c00000009
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x3000000ddbc0001
dataLength = 186
numChildren = 0

   控制器遍历这些Follower副本分区，并确定谁应该成为新Leader分区，然后向所有包含新Leader分区和现有
Follower的 broker 发送请求。该请求消息包含了谁是新Leader副本分区以及谁是Follower副本分区的信息。随后，
新Leader分区开始处理来自生产者和消费者的请求，而跟随者开始从新Leader副本分区消费消息。
   当控制器发现一个 broker 加⼊集群时，它会使用 broker ID 来检查新加⼊的 broker 是否包含现有分区的副本。
如果有，控制器就把变更通知发送给新加入的 broker 和其他 broker，新 broker上的副本分区开始从Leader分区那里
消费消息，与Leader分区保持同步。
   结论：
   (1).Kafka使用 Zookeeper 的分布式锁选举控制器，并在节点加入集群或退出集群时通知控制器。
   (2).控制器负责在节点加入或离开集群时进行分区Leader选举。
   (3).控制器使用epoch来避免“脑裂”。“脑裂”是指两个节点同时认为自己是当前的控制器。
   